package com.g.ocache.utils;

import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 
 * @author Jingqi Xu
 */
public class CallableExecutor<V> {
	//
	private final AtomicBoolean cachable;
	private final ConcurrentHashMap<Object, FutureTask<V>> tasks;
	
	/**
	 * 
	 */
	public CallableExecutor() {
		this(false);
	}
	
	public CallableExecutor(boolean cachable) {
		this.cachable = new AtomicBoolean(cachable);
		this.tasks = new ConcurrentHashMap<Object, FutureTask<V>>();
	}
	
	/**
	 * 
	 */
	public boolean isCachable() {
		return cachable.get();
	}
	
	public void setCachable(boolean cachable) {
		this.cachable.set(cachable);
	}
	
	/**
	 * 
	 * @param id
	 * @param callable
	 * @return null if current thread was interrupted
	 * @throws Exception
	 */
	public V execute(Object id, Callable<V> callable) throws Exception {
		//
		FutureTask<V> task = tasks.get(id);
		if(task == null) {
			task = new FutureTask<V>(callable);
			FutureTask<V> existing = tasks.putIfAbsent(id, task);
			if(existing == null) {
				task.run();
			} else {
				task = existing;
			}
		}
		
		//
		try {
			return task.get();
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			return null;
		} catch (ExecutionException e) {
			if(e.getCause() instanceof Error) {
				throw (Error)e.getCause();
			} else {
				throw (Exception)e.getCause();
			} 
		} finally {
			//
			if(!isCachable()) {
				tasks.remove(id, task);
			}
		}
	}
}
